package com.denghq.projectbuilder.component.basedatacache.builder;

import com.denghq.projectbuilder.component.basedatacache.autoconfigure.BaseDataCacheProperties;
import com.denghq.projectbuilder.component.basedatacache.cache.ICacheDomain;
import com.denghq.projectbuilder.component.msgbus.bus.impl.RabbitMqMsgBus;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.SmartLifecycle;
import org.springframework.util.CollectionUtils;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Data
@Slf4j
public class BaseDataCacheHolder implements SmartLifecycle {

    private List<ICacheDomain> cacheList;

    private BaseDataCacheProperties properties;

    private ExecutorService executor;

    private final AtomicBoolean running;
    private Random r = new Random();

    /**
     * 消息总线，用来订阅消息
     */
    private final RabbitMqMsgBus eventBus;

    private BaseDataCacheHolder(BaseDataCacheProperties properties, List<ICacheDomain> cacheList, RabbitMqMsgBus eventBus) {
        this.eventBus = eventBus;
        this.running = new AtomicBoolean(false);
        this.cacheList = cacheList;
        this.properties = properties;
        this.executor = Executors.newFixedThreadPool(properties.getMaxThreadNum());

    }


    private void mqRefresh() {

        //List<CacheDomain> cacheList = Lists.newArrayList();
        //List<String> topicList = Lists.newArrayList();
        this.cacheList.forEach(cache -> {
            if (StringUtils.isNotBlank(cache.getTopicName())) {
                int retryCount = (properties.getFailRetry() == null || properties.getFailRetry() < 1) ? 1 : properties.getFailRetry();
                UpdateCacheAction action = new UpdateCacheAction(cache, retryCount, properties.getFailRetryMaxSecond(), properties.getFailRetryMinSecond(), executor);
                Arrays.stream(cache.getTopicName().split(",")).forEach(topic -> eventBus.subscribe(topic, action));
            }
        });

    }

    private void timedRefresh() {
        executor.execute(() -> {
            while (true) {
                try {
                    //随机睡一会，避免大量请求同时发出,出现请求热点
                    TimeUnit.MILLISECONDS.sleep(r.nextInt(500));
                    loadCache();
                    TimeUnit.MINUTES.sleep(properties.getCacheTimeOutMin());
                } catch (InterruptedException e) {
                    log.info("获取基础数据缓存休眠被中断:{}", e.getMessage());
                    break;
                }
            }
        });
    }

    private void loadCache() {
        cacheList.forEach(cache -> {
            int retryCount = (properties.getFailRetry() == null || properties.getFailRetry() < 1) ? 1 : properties.getFailRetry();
            executor.execute(() -> {
                int i = 0;
                while (i <= retryCount) {
                    try {
                        int count = cache.initLoad();
                        cache.setLoadSuccess(true);
                        log.info("获取基础数据缓存-【{} {}】数据（{}）条", cache.getCacheKey(), cache.getCacheName(), count);
                        break;
                    } catch (Exception e) {
                        i++;
                        Random random = new Random();
                        Integer second = random.nextInt(properties.getFailRetryMaxSecond() - properties.getFailRetryMinSecond()) + properties.getFailRetryMinSecond();
                        log.error("获取基础数据缓存-【{} {}】数据发生异常,将在{}s后进行第{}次重试", cache.getCacheKey(), cache.getCacheName(), second, i);
                        try {
                            TimeUnit.SECONDS.sleep(second);
                        } catch (InterruptedException el) {
                            log.info("获取基础数据缓存-【{} {}】数据休眠被中断", cache.getCacheKey(), cache.getCacheName(), el.getMessage());
                            break;
                        }
                    }
                }
            });
        });
    }


    public <T> ICacheDomain<T> getCache(String cacheKey) {
        if (!CollectionUtils.isEmpty(cacheList)) {
            ICacheDomain<T> cache = cacheList.stream().filter(cacheInfo ->
                    cacheInfo.getCacheKey().equals(cacheKey)
            ).findFirst().orElse(null);
            if (cache != null) {
                if (cache.getLoadSuccess()) {
                    return cache;
                } else {
                    cache.initLoad();
                    cache.setLoadSuccess(true);
                    return cache;
                }
            }
        }
        return null;
    }

    public List<ICacheDomain> getAllCache() {
        if (!CollectionUtils.isEmpty(cacheList)) {
            cacheList.forEach(c -> this.getCache(c.getCacheKey()));
        }
        return cacheList;
    }

    private volatile static BaseDataCacheHolder instance = null;

    //单例模式
    public static BaseDataCacheHolder getInstance(BaseDataCacheProperties properties, List<ICacheDomain> cacheList, RabbitMqMsgBus eventBus) {
        //double check
        if (instance == null) {
            synchronized (BaseDataCacheHolder.class) {
                if (instance == null) {
                    instance = new BaseDataCacheHolder(properties, cacheList, eventBus);
                }
            }
        }
        return instance;
    }

    public void init() {
        if (!CollectionUtils.isEmpty(cacheList)) {
            //定时刷新缓存数据
            timedRefresh();
            //监听MQ消息更新缓存数据
            mqRefresh();
        }
    }

    @Override
    public boolean isAutoStartup() {
        return true;
    }

    @Override
    public void start() {
        if (this.running.compareAndSet(false, true)) {
            this.init();
        }
    }

    @Override
    public void stop(Runnable callback) {
        try {
            this.stop();
        } catch (Exception e) {

        }
        callback.run();
    }

    @Override
    public void stop() {
        if (this.running.compareAndSet(true, false)) {
            executor.shutdown();
        }
    }

    @Override
    public boolean isRunning() {
        return this.running.get();
    }

    @Override
    public int getPhase() {
        return 0;
    }
}
